#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <iostream>
#include <vector>
#include <cstdint>
#include <cassert>
#include <cstring>
#include <sys/socket.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <ctime>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <sys/eventfd.h>
#include <thread>
#include <sys/timerfd.h>
#include <typeinfo>
#include <condition_variable>
#include <pthread.h>
#include <signal.h>
#define BUFFER_DEFAULT_SIZE  1024
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF
#define LOG(level,format,...) do{\
        if (level < LOG_LEVEL) break;\
        const time_t t = time(nullptr);\
        struct tm* ltm = localtime(&t);\
        char tmp[32] = {0};\
        strftime(tmp,31,"%H:%M:%S",ltm);\
        fprintf(stdout,"[%p %s:%s:%d] " format "\n",(void*)pthread_self(),tmp,__FILE__,__LINE__,##__VA_ARGS__);\
    }while(0)
#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)
class Buffer{
private:
    std::vector<char> _buffer; // 使用vector管理
    size_t _read_idx; // 读偏移量
    size_t _write_idx; //
public:
    Buffer():_buffer(BUFFER_DEFAULT_SIZE),_read_idx(0),_write_idx(0) {}
    // 获取当前的写入起始地址
    char* WritePosition(){
        // &_buffer + _write_idx
        return beginAddress() + _write_idx;
    }
    // 获取当前的读取起始地址
    char* ReadPosition(){
        return beginAddress() + _read_idx;
    }
    // 获取写之后的空闲空间
    size_t TailIdleSize() {return _buffer.size() - _write_idx;}
    // 获取读之前的空闲空间
    size_t HeadIdleSize() {return _read_idx;}
    // 可获取的数据大小
    size_t ReadAbleSize() {return _write_idx - _read_idx;}
    // 读偏移更新
    void MoveReadOffset(size_t sz) {
        assert(sz <= ReadAbleSize());
        if(sz == 0) return;
        _read_idx += sz;
    }
    // 写偏移更新
    void MoveWriteOffset(size_t sz) {
        assert(sz <= TailIdleSize());
        _write_idx += sz;
    };
    // 确保可写空间足够
    void EnsureWriteCapacity(size_t sz){
        if (TailIdleSize() >= sz) return;
        // 尾部空间不够
        if(HeadIdleSize() + TailIdleSize() >= sz){
            size_t readsz = ReadAbleSize(); // 把当前的可读数据长度保存
            // strncmp(&_buffer[0],(char*)ReadPosition(),readsz);
            std::copy(ReadPosition(),ReadPosition() + readsz,beginAddress()); // 把拷贝数据保存到起始位置
            _read_idx = 0;// 读置零
            _write_idx = readsz;// 写偏移量
        }else{
            _buffer.resize(_write_idx + sz);
        }
    }
    // 写入数据
    void Write(const void* data,const size_t len){
        if(len <= 0) return;
        // 足够的空间  + 拷贝数据 
        EnsureWriteCapacity(len);
        // strncpy(&_buffer[_read_idx],(char*)data,len);
        const char* d = static_cast<const char*>(data);
        std::copy(d,d+len,WritePosition());
        // _write_idx += len;
    }
    void WriteAndPush(const void* data,const size_t len){
        Write(data,len);
        MoveWriteOffset(len);
    }
    // 重载string版本
    void WriteString(const std::string& data){
        return Write(data.c_str(),data.size());
    }
    void WriteStringAndPush(const std::string& data){
        WriteString(data);
        MoveWriteOffset(data.size());
    }
    void WriteBuffer(Buffer& data){
        return Write(data.ReadPosition(),data.ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer& data) {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }
    // 读取空间
    void Read(void* buff,size_t len){
        // 获取的数据大小 <= 可读数据大小
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(),ReadPosition()+len,(char*)buff);
        // _read_idx += len;
    }
    void ReadAndPop(void* buf,size_t len){
        Read(buf,len);
        MoveReadOffset(len);
    }
    std::string ReadAsString(size_t len){
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0],len);
        return str;
    }
    std::string ReadAsStringAndPop(size_t len){
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    //清空缓冲区
    void CleanBuffer() {_read_idx = _write_idx = 0;}
    const void* FindCRLF(){
        const void* res = memchr(ReadPosition(),'\n',ReadAbleSize());
        return res;
    }
    std::string GetLine(){
        char* pos = static_cast<char*>(const_cast<void*>(FindCRLF()));
        if(pos == nullptr){
            return "";
        }
        return ReadAsString(pos - ReadPosition() + 1); // 顺便取出换行字符
    }
    std::string GetLineAndPop(){
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
private:
    char* beginAddress(){
        return &(*_buffer.begin()); // &_buffer[0];
    }
};

#define MAX_LISTEN 1024
class Socket{
public:
    Socket():_socketfd(-1) {}
    Socket(int fd):_socketfd(fd) {}
    ~Socket() {Close();}
    int Fd(){
        return _socketfd;
    }
    // 创建
    bool Create(){
        _socketfd = socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
        if (_socketfd < 0)
        {
            ERR_LOG("create socket err");
            return false;
        }
        return true;
    }
    // bind
    bool Bind(const std::string& ip,int port){
        struct sockaddr_in addr;
        addr.sin_family=AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = bind(_socketfd,(struct sockaddr*)&addr,len);
        if(ret < 0){
            ERR_LOG("bind err");
            return false;
        }
        return true;
    }
    // listen
    bool Listen(int backlog=MAX_LISTEN){
        int ret = listen(_socketfd,backlog);
        if(ret < 0){
            ERR_LOG("LISTEN ERR");
            return false;
        }
        return true;
    }
    // 发起连接
    bool Connect(const std::string& ip,int port){
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_socketfd,(struct sockaddr*)&addr,len);
        if(ret < 0){
            ERR_LOG("CONNECT SERVER ERR");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept(){
        int ret = accept(_socketfd,nullptr,nullptr);
        if(ret < 0){
            ERR_LOG("ACCEPT ERR");
            return -1;
        }
        return ret;
    }
    // recv
    ssize_t Recv(void* buf,size_t len,int flag=0){
        ssize_t sz = recv(_socketfd,buf,len,flag);
        if(sz <= 0){
            if(errno==EAGAIN || errno==EINTR){
                // EAGAIN 缓冲区没数据 
                //EINTR socket阻塞等待被打断了
                return 0;
            }
            ERR_LOG("RECV ERR");
            return -1;
        }
        return sz;
    }
    ssize_t RecvNonBlock(void* buf,size_t len){
        return Recv(buf,len,MSG_DONTWAIT);// MSG_DONTWAIT 非阻塞
    }
    // send
    ssize_t Send(void* buf,size_t len,int flag=0){
        ssize_t ret = send(_socketfd,buf,len,flag);
        if(ret < 0){
            ERR_LOG("SEND ERR");
            return -1;
            if(errno==EAGAIN || errno==EINTR){
                return 0;
            }
        }
        return ret;
    }
    ssize_t SendNonBlock(void* buf,size_t len){
        if(len == 0) return 0;
        return Send(buf,len,MSG_DONTWAIT);
    }
    // 关闭连接
    void Close(){
        if(_socketfd != -1){
            close(_socketfd);
            _socketfd = -1;
        }
    }
    //client-link
    bool CreateClient(const std::string& ip,int port){
        if(!Create()) return false;
        if (!Connect(ip,port)) return false;
        return true;
    }
    // server-link
    bool CreateServer(int port,const std::string& ip="0.0.0.0",bool flag = false){
        if(!Create()) return false;
        if(!Bind(ip,port)) return false;
        if(!Listen()) return false;
        if(flag)
            NonBlock();
        ReuseAddr();
        return true;
    }
    // set...opt
    void ReuseAddr(){
        int opt = 1;
        // int setsockopt(int __fd, int __level, int __optname,const void *__optval, socklen_t __optlen)
        setsockopt(_socketfd,SOL_SOCKET,SO_REUSEADDR,(void*)&opt,sizeof(opt)); // 地址复用
        opt = 1;
        setsockopt(_socketfd,SOL_SOCKET,SO_REUSEPORT,(void*)&opt,sizeof(opt)); // 端口复用
    }
    // 非阻塞式选项
    void NonBlock(){
        int flag = fcntl(_socketfd,F_GETFL,0);
        fcntl(_socketfd,F_SETFL,flag|O_NONBLOCK);
    }
private:
    int _socketfd; 
};
/// @brief Channel //////////////////
class Poller;
class EventLoop;
using EventCallback = std::function<void()>;
class Channel{
public:
    Channel(EventLoop* loop,int fd):_fd(fd),_events(0),_revents(0),_loop(loop) {}
    int Fd() {return _fd;}
    // 可读写
    bool ReadAble() {return _events & EPOLLIN;}
    bool WriteAble() {return _events & EPOLLOUT;}
    // 启动选项 , 在eventLoop中调用
    void EnableRead() {_events |= EPOLLIN; Update();}
    void EnableWrite() {_events |= EPOLLOUT; Update();}
    void DisableRead() {_events &= ~EPOLLIN; Update(); }
    void DisableWrite() {_events &= ~EPOLLOUT; Update();}
    void DisableAll() {_events = 0; Update();}
    // 移除监控 
    void Remove();
    void Update();
    uint32_t Events() {return _events;}
    // 绑定回调
    void SetREvents(uint32_t events) { _revents = events;}
    void SetReadCallback(const EventCallback& cb) {_read_callback = cb;}
    void SetwriteCallback(const EventCallback& cb) {_write_callback = cb;}
    void SeterrorCallback(const EventCallback& cb) {_error_callback = cb;}
    void SetcloseCallback(const EventCallback& cb) {_close_callback = cb;}
    void SeteventCallback(const EventCallback& cb) {_event_callback = cb;}
    // 判断事件类型，调用对应的回调
    void HandleEvent(){
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP || (_revents & EPOLLPRI))){
            // 任何事件调用接口
            if (_read_callback) _read_callback();
        }
        if(_revents & EPOLLOUT){
            // 任何事件调用接口
            if(_event_callback) _event_callback();
            if(_write_callback) _write_callback();
        }
        else if(_revents & EPOLLERR)
            if(_error_callback) _error_callback();
        else if(_revents & EPOLLHUP)
            if(_close_callback) _close_callback();
        if(_event_callback) _event_callback(); 
    } 
private:
    int _fd; 
    EventLoop *_loop;
    uint32_t _events; // 监控的事件
    uint32_t _revents; // 触发的事件 全用位表示 

    EventCallback _read_callback;
    EventCallback _write_callback;
    EventCallback _error_callback; // 错误事件
    EventCallback _close_callback; // 连接断开
    EventCallback _event_callback; // 任意事件,刷新连接活跃
};
#define MAX_EPOLLEVENTS 1024
class Poller{
public:
    Poller(){
        _epfd = epoll_create(5);
        if(_epfd < 0){
            ERR_LOG("EPOLL CREATE ERR");
            abort();
        }
    }
    void UpdateEvents(Channel* channel){
        if(!HasChannel(channel)){
            Update(channel,EPOLL_CTL_ADD);
            _channels.insert({channel->Fd(),channel});
        }
        else
            Update(channel,EPOLL_CTL_MOD);

    }
    void RemoveEvent(Channel* channel){
        // 将hash中的信息也删掉
        if(HasChannel(channel))
            _channels.erase(channel->Fd());
        Update(channel,EPOLL_CTL_DEL);
    }
    // 监控，返回活跃连接
    void Poll(std::vector<Channel*> *actives){
        int nfds = epoll_wait(_epfd,_evs,MAX_EPOLLEVENTS,-1);
        if(nfds < 0){
            if(errno == EINTR) return;
            ERR_LOG("EPOLL WAIT ERR:%s\n",strerror(errno));
            abort();
        }
        for(int i= 0;i < nfds;i++){
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetREvents(_evs[i].events); // 设置就绪的文件
            actives->push_back(it->second);
        }
    }
private:
    // epoll_control
    void Update(Channel* channel,int opt){
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();

        int ret = epoll_ctl(_epfd,opt,fd,&ev);
        if(ret < 0){
            ERR_LOG("EPOLL_CTL ERR");
            abort(); // 退出程序
        }
        return;
    }
    // 判断是否有相关的事件的监控
    bool HasChannel(Channel* channel){
        return _channels.find(channel->Fd()) !=_channels.end();
    }
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    std::unordered_map<int,Channel*> _channels;
};


///////


using TaskFunc = std::function<void()>;
using RealeaseFunc = std::function<void()>;
class TimerTask{ // 对象什么时候执行什么任务
private:
    uint64_t _id; // 定时器对象id
    uint32_t _timeout; // 超时时间
    TaskFunc _callback; // 定时器要执行的定时任务
    bool _canceled;// false 没有取消
    RealeaseFunc _release; // 删除TimerWheel中的定时器对象信息
public:
    TimerTask(uint64_t id,uint32_t timeout,const TaskFunc& callback,bool cancel = false):_id(id),_timeout(timeout),_callback(callback),_canceled(cancel){}
    ~TimerTask(){
        if(!_canceled)
            _callback();
        _release();
    }
    inline void ChangeCancel(){_canceled = !_canceled;}
    void SetRelease(const RealeaseFunc& cb){_release = cb;}
    inline uint32_t GetTimeout(){return _timeout;}
};
using WeakTask = std::weak_ptr<TimerTask>;
using PtrTask = std::shared_ptr<TimerTask>;
class TimerWheel{
public:
    TimerWheel(EventLoop* loop):_tick(0),_capacity(60),_wheel(_capacity),_loop(loop),_timerfd(CreateTimerFd()),_timer_channel(new Channel(_loop,_timerfd))
    {
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
        _timer_channel->EnableRead();
    }
    // Add 定时器中有个timers成员，定时器的信息操作可能在多线程中操作，因此需要考虑线程安全
    // 不想枷锁，就需要定期的所有操作放在一个线程中进行
    void TimerAdd(uint64_t id,uint32_t timeout,const TaskFunc& callback);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    
    // 存在线程安全问题
    bool HasTimer(uint64_t id){
        return _timers.find(id) != _timers.end();
    }
private:
    void TimerAddInLoop(uint64_t id,uint32_t timeout,const TaskFunc& callback);
    void TimerRefreshInLoop(uint64_t id);
    void TimerCancelInLoop(uint64_t id);
    void RemoveTask(uint64_t id){
        if(_timers.find(id)!=_timers.end()){
            _timers.erase(id);
        }
    }
    static int CreateTimerFd(){
        // int timerfd_create(int clockid,int flags);
    int timerfd = timerfd_create(CLOCK_MONOTONIC,0);
    if(timerfd < 0){
        ERR_LOG("timerfd create err");
        abort();
    }
    // int timerfd_settime(int fd,int flags,struct itimerspec* new,struct itimerspec* old); 
    struct itimerspec itime;
    // 设置第一次超时时间
    itime.it_value.tv_sec = 1;
    itime.it_value.tv_nsec = 0;
    // 超时时间间隔
    // itime.it_interval.tv_sec = 1;
    // itime.it_interval.tv_nsec = 0;
    itime.it_interval = {1,0};
    timerfd_settime(timerfd,0,&itime,nullptr);
    return timerfd;
    }
    int ReadTimerFd(){
        uint64_t times = 0;
        // 有可能因为其他的描述符的事件处理花费的事件比较长，然后再定时器描述符处理时间的时候，有可能就已经超时了很多次
        // read读取的数据times就是上一次read之后的超时次数
        int ret = read(_timerfd,&times,sizeof(uint64_t));
        if(ret < 0){
            ERR_LOG("TIMERFD READ ERR");
            abort();
        }
        return times;
    }
    void RunTimerTask();
    void OnTime(){
        // 根据实际超时的次数，执行对应的超时任务
        int times = ReadTimerFd();
        for(int i = 0 ;i < times;i++){
            RunTimerTask();
        }

    }
private:
    int _tick; // 指哪里销毁哪里
    int _capacity;
    std::vector<std::vector<PtrTask>> _wheel;
    std::unordered_map<uint64_t,WeakTask> _timers;
    EventLoop* _loop;
    int _timerfd; // 定时器描述符
    std::unique_ptr<Channel> _timer_channel;
};

void TimerWheel::TimerAddInLoop(uint64_t id,uint32_t timeout,const TaskFunc& callback){
    PtrTask pt(new TimerTask(id,timeout,callback));
    pt->SetRelease(std::bind(&TimerWheel::RemoveTask,this,id));
    _timers.insert({id,WeakTask(pt)});
    // 添加到轮子中
    _wheel[(timeout+_tick)%_capacity].emplace_back(pt);
}
void TimerWheel::TimerRefreshInLoop(uint64_t id){
    // 通过保存的定时器对象的weakptr构造一个shared_ptr中
    if(_timers.find(id) == _timers.end()){
        return;
    }
    auto WeakPtr = _timers.at(id);
    PtrTask pt = WeakPtr.lock(); // 获取weak_ptr管理的shared_ptr
    int delay = pt->GetTimeout();
    _wheel[(_tick + delay)%_capacity].emplace_back(pt);
}
void TimerWheel::RunTimerTask(){
    _tick = (_tick + 1) % _capacity;
    //走到哪里释放哪里
    _wheel[_tick].clear();
}
void TimerWheel::TimerCancelInLoop(uint64_t id){
    if(_timers.find(id) == _timers.end()) return;
    auto pt = _timers.at(id).lock(); // 拿到shared_ptr
    if(pt) pt->ChangeCancel(); 
}


using Functor = std::function<void()>;
class EventLoop{
public:
    EventLoop():_thread_id(std::this_thread::get_id()),
                _event_fd(CreateEventFd()),
                _event_channel(new Channel(this,_event_fd)),
                _timer_wheel(this)
                {
                    // 给eventfd添加回调函数，读取eventfd时间通知次数
                    _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
                    _event_channel->EnableRead();
                }
    // 判断要执行的任务是否处于当前的线程中，是执行 否压入队列
    void RunInLoop(const Functor& cb){
        if(IsInLoop()){
            return cb();
        }else{
            return QueueInLoop(cb);
        }
    }
    // 将任务加入任务池
    void QueueInLoop(const Functor& cb){
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        // 唤醒有可能没有事件就绪，而导致的epoll阻塞
        WeakUpEventFd();
    }
    // 判断当前线程是否为EventLoop对应的线程
    bool IsInLoop(){
        return _thread_id == std::this_thread::get_id();
    }
    void AssertInLoop(){
        assert(_thread_id == std::this_thread::get_id());
    }
    void UpdateEvent(Channel* channel){
        return _poller.UpdateEvents(channel);
    }
    void RemoveEvent(Channel* channel){
        return _poller.RemoveEvent(channel);
    }
    // 事件监控->就绪事件处理->执行任务
    void Start(){
        while(true){
            std::vector<Channel*> actives;
            _poller.Poll(&actives);
            for(auto& channel:actives){
                channel->HandleEvent();
            }
            RunAllTask();
        }
    }

    void TimerAdd(uint64_t id,uint32_t timeout,const TaskFunc& cb){
        return _timer_wheel.TimerAdd(id,timeout,cb);
    }
    void TimerRefresh(uint64_t id){
        return _timer_wheel.TimerRefresh(id);
    }
    void TimerCancel(uint64_t id){
        return _timer_wheel.TimerCancel(id);
    }
    bool HasTimer(uint64_t id){
        return _timer_wheel.HasTimer(id);
    }
private:
    // 执行任务池所有任务 
    void RunAllTask(){
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);
        }
        for(auto& f : functor){
            f();
        }
        return;
    }
    static int CreateEventFd(){
       int efd = eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);
        if(efd < 0){
            ERR_LOG("_EVENT_FD CREATE ERR");
            abort();
        }
        return efd;
    }
    void ReadEventFd(){
        uint64_t res = 0;
        int ret = read(_event_fd,&res,sizeof(res));
        if(ret < 0){
            if(errno == EINTR || errno == EAGAIN) {
                return;
            }
            ERR_LOG("READ EVENTFD ERR");
            abort();
        }
        return;
    }
    // 唤醒有可能没有事件就绪，而导致的epoll阻塞
    // 给eventfd写入一个数据，eventfd就会触发可读事件
    void WeakUpEventFd(){
        uint64_t res = 0;
        int ret = write(_event_fd,&res,sizeof(res));
        if(ret < 0){
            if(errno == EINTR || errno==EAGAIN){
                return;
            }
            ERR_LOG("EVENTFD WRITE ERR");
            abort();
        }
        return;
    }
private:
    int _event_fd; // event_fd唤醒IO事件导致的阻塞
    Poller _poller; // 事件监控
    std::unique_ptr<Channel> _event_channel;
    std::vector<Functor> _tasks; // 任务池
    std::mutex _mutex; // 保证任务池的安全
    std::thread::id _thread_id; // 获取当前线程的id
    TimerWheel _timer_wheel;//定时器模块
};


class LoopThread{
private:
    // 用户实现那_loop获取同步关系，避免线程创建，_loop还未实例化之前获取_loop
    std::mutex _mutex; // 
    std::condition_variable _cond;
    EventLoop *_loop; // 对象指针，对象需要在线程内实例化
    std::thread _thread; // eventloop 对应的线程
private:
    void ThreadEntry() // 实例化EventLoop 并运行EventLoop模块的功能
    {
        EventLoop loop; // 线程id 就是当前线程的id
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }
public:
    // 创建线程，设定线程入口函数
    LoopThread():_loop(nullptr),_thread(std::thread(&LoopThread::ThreadEntry,this)) {}
    EventLoop* GetLoop() // 返回当前线程关联的EventLoop对象的指针
    {
        EventLoop* loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex); // 枷锁
            _cond.wait(lock,[&](){return _loop != nullptr;}); // loop为null，就一直阻塞
            loop = _loop;
        }
        return loop;
    }
};


class LoopThreadPool{
private:
    int _thread_cout;//从属线程的数量
    int _next_idx;
    EventLoop* _baseloop; // 主Reactor，从属为0，所有的操作都在baseloop中进行
    std::vector<LoopThread*> _threads;// 保存所有的LoopThread对象
    std::vector<EventLoop*> _loops;
public:
    LoopThreadPool(EventLoop* base_loop):_baseloop(base_loop),_thread_cout(0),_next_idx(0) {}
    void SetThreadCount(int count) // 设置线程数量
    {
        _thread_cout = count;
    }
    void Create()// 创建所有的从属线程
    {
        if(_thread_cout > 0){
            _threads.resize(_thread_cout);
            _loops.resize(_thread_cout);
            for(int i = 0; i < _thread_cout;i++){
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }
    EventLoop* NextLoop() // 分配线程
    {
        if(_thread_cout == 0) {
            return _baseloop;
        }
        _next_idx = (_next_idx + 1) % _thread_cout;
        return _loops[_next_idx];
    }
};



class Any{
private:
    class hoder{
    public:
        virtual ~hoder() {}
        virtual const std::type_info& type()=0; // 获取类型
        virtual hoder* clone() = 0; // 拷贝
    };
    template<class T>
    class placehoder:public hoder{
    public:
    friend Any;
        placehoder(const T& val):_val(val){}
        virtual const std::type_info& type(){return typeid(_val);} // 获取类型
        virtual hoder* clone(){ // 拷贝
            return new placehoder(_val);
        }
    private:
        T _val;
    };
    hoder* _content;
public:
    Any():_content(nullptr) {}
    template <class T>
    Any(const T& val):_content(new placehoder<T>(val)) {}
    Any(const Any& other):_content(other._content != nullptr ? other._content->clone():nullptr) {}
    ~Any() {
        if(_content)
            delete _content;
    }
    Any& swap(Any& other){
        std::swap(_content,other._content);
        return *this;
    }
    template<class T>
    T* get(){ // 返回子类对象保存的数据的指针
    // 获取的数据类型，必须和保存的数据类型一致
        if(typeid(T) == _content->type())
            // return &((placehoder<T>*)_content)->_val;
            return &(dynamic_cast<placehoder<T>*>(_content))->_val;
        return nullptr;
    }
    template<class T>
    Any& operator=(const T& val){ // 赋值重载
        Any(val).swap(*this);
        return *this;
    }
    Any& operator=(const Any& other){
        Any(other).swap(*this);
        return *this;
    }
    
};
// 连接关闭  连接成功--待处理  连接建立完成--已连接 断开连接中
class Connnection;
typedef enum {DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING} ConnStatus;
using PtrConnection = std::shared_ptr<Connnection>;

class Connnection:public std::enable_shared_from_this<Connnection> {
private:
    uint64_t _conn_id; // 连接的唯一id，便于连接和管理
    // uint64_t _timer_fd; //定时器id，必须是唯一的，简化使用conn_id作为定时器id
    int _sockfd; // 连接关联的文件描述符
    bool _enable_inactive_release; // 连接是否启动非活跃销毁的判断标值，默认是false
    EventLoop *_loop; // 连接所关联的EventLoop
    ConnStatus _status; // 标记当前的状态
    Socket _socket;
    Channel _channel; // 事件管理
    Buffer _in_buffer; // socket读取的放入输入缓冲区
    Buffer _out_buffer; // 存放发送给对端的数据
    Any _context; // 请求接收处理上下文
// 服务器模块设置 ，这几个都是组件使用者使用的
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&,Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    ClosedCallback _server_closed_callback; // 组件内的连接关闭回调，因为服务器是进行连接管理的

private:
    // 五个channel的事件回调函数
    //读取接收socket数据，放到缓冲区中，然后调用_message_callback
    void HandleRead(){
        // 1.接收socket数据放入缓冲区
        char buf[65536] = {0};
        ssize_t ret = _socket.RecvNonBlock(buf,65535);
        if(ret < 0){
            // 出错了，不能直接关闭连接
            return ShutdownInLoop();
        }
        _in_buffer.WriteAndPush(buf,ret);
        // 2.调用message_callback 进行业务处理
        if(_in_buffer.ReadAbleSize() > 0){
            // shared_from_this 从当前对象自身 获取自身的shared_ptr管理对象
            return _message_callback(shared_from_this(),&_in_buffer);
        }
    }
    void HandleWrite() // 数据放入发送缓冲区
    {
        // _out_buffer 保存的数据就是要发送的数据
        ssize_t ret = _socket.SendNonBlock(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());
        if(ret < 0){
            // 发送失败
            if(_in_buffer.ReadAbleSize() > 0){
            // shared_from_this 从当前对象自身 获取自身的shared_ptr管理对象
                _message_callback(shared_from_this(),&_in_buffer);
            }
            return Release(); // 实际的关闭释放操作
        }
        _out_buffer.MoveReadOffset(ret); // 将偏移量向后偏移
        if(_out_buffer.ReadAbleSize() == 0){
           _channel.DisableWrite(); // 无事件待发送
        }
        // 如果当前是连接待关闭状态 则有数据就发送 无数据 就释放
        if(_status == DISCONNECTING && _out_buffer.ReadAbleSize() == 0){
            return Release();
        }
        return;
    }
    void HandleClose() // 触发挂断事件
    {
        // 一但连接挂断，套接字什么都做不了，有待处理的数据处理一下
        if(_in_buffer.ReadAbleSize() > 0){
            _message_callback(shared_from_this(),&_in_buffer);
        }
        return Release();
    }
    void HandleError() // 触发出错事件
    {
        HandleClose();
    }
    void HandleEvent() // 任意事件 1.延迟销毁任务 2.调用用户组件的任意事件回调
    {
        if(_enable_inactive_release)
            _loop->TimerRefresh(_conn_id);
        if(_event_callback)
            _event_callback(shared_from_this());
    }
    void EstablishedInLoop() // 连接获取之后，所处的状态下要进行的各种设置，channel设置各种事件回调，启动读监控
    {
        assert(_status == CONNECTING); // 当前状态必须是半连接状态
        // 1.修改连接状态
        _status = CONNECTED;
        // 一旦启动读事件监控，有可能就直接出发了读事件，如果这个时候启动了非活跃链接销毁
        // 2.启动读写事件
        _channel.EnableRead();
        if(_connected_callback) _connected_callback(shared_from_this());
        // 3.调用回调函数
    }
    void SendInLoop(Buffer& buf) // 不是实际的发送接口，是发送缓冲区中,启动写事件监控
    {
        if(_status == DISCONNECTED) return;
        _out_buffer.WriteBufferAndPush(buf);
        if(_channel.WriteAble() == false){
            _channel.EnableWrite();
        }
    }
    // 关闭操作不是实际的连接释放操作，需要判断还有没有数据待处理
    void ShutdownInLoop()
    {
        // 设置连接为半关闭
        _status = DISCONNECTING;
        if(_in_buffer.ReadAbleSize() > 0){
            if(_message_callback)
                _message_callback(shared_from_this(),&_in_buffer);
        }
        // 要么是写入数据出错关闭，要么是没有待发送的数据，直接关闭
        if(_out_buffer.ReadAbleSize() > 0) {
            if(_channel.WriteAble() == false){
            _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadAbleSize() == 0){
            Release();
        }

    }
    void RealseInLoop() // 实际的释放接口
    {
        // 1.修改连接状态
        _status = DISCONNECTED;
        // 2.移除连接的事件监控
        _channel.Remove();
        //3. 关闭描述符
        _socket.Close();
        // 4.如果当前的定时器中还有定时销毁任务，需要取消
        if(_loop->HasTimer(_conn_id)) CancelInactiveRealseInLoop();
        // 5.调用关闭回调函数，避免先移除服务器的，导致connection被释放，再去处理会出错，所以先调用用户的回调函数
        if(_closed_callback) _closed_callback(shared_from_this());
        // 移除服务器内部管理的连接信息
        if(_server_closed_callback) _server_closed_callback(shared_from_this());
    }
    // 启动非活跃超时链接释放规则
    void EnableInactiveRealseInLoop(int sec){
        // 1. 将判断标值 _enable_inactiev_release = true
        _enable_inactive_release = true;
        // 2.如果当前定时销毁任务存在 就刷新任务 
        if(_loop->HasTimer(_conn_id))
            _loop->TimerRefresh(_conn_id);
        // 3.不存在就加入
        _loop->TimerAdd(_conn_id,sec,std::bind(&Connnection::Release,this));
    }
    void CancelInactiveRealseInLoop(){
        _enable_inactive_release = false;
        if(_loop->HasTimer(_conn_id))
            _loop->TimerCancel(_conn_id);
    }
    void UpgradeInLoop(const ConnectedCallback& conn,const MessageCallback& msg,const ClosedCallback& closed,
    const AnyEventCallback& event,const Any &context){
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }
public:
    Connnection(EventLoop* loop,uint64_t conn_id,int sockfd)
            :_loop(loop),_conn_id(conn_id),_sockfd(sockfd),
            _enable_inactive_release(false),_status(CONNECTING),_socket(_sockfd),
            _channel(_loop,_sockfd)
            {
                _channel.SetcloseCallback(std::bind(&Connnection::HandleClose,this));
                _channel.SeteventCallback(std::bind(&Connnection::HandleEvent,this));
                _channel.SetReadCallback(std::bind(&Connnection::HandleRead,this));
                _channel.SetwriteCallback(std::bind(&Connnection::HandleWrite,this));
                _channel.SeterrorCallback(std::bind(&Connnection::HandleError,this));
            }
    ~Connnection() { DBG_LOG("RELEASE CONNECTION:%p",this); }
    int Fd() { return _sockfd; }
    int Id() { return _conn_id; }
    bool Connected() // 是否处于连接状态
    {
        return _status == CONNECTED;
    }
    void SetContext(const Any& context) // 设置上下文 -- 链接建立完成时
    {
        _context = context;
    }
    Any* GetContext()// 获取上下文
    {
        return &_context;
    }
    void SetConnectedCallback(const ConnectedCallback& cb){
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb){
        _message_callback=  cb;
    }
    void SetClosedCallback(const ClosedCallback& cb) {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback& cb){
        _event_callback = cb;
    }
    void SetSvrClosedCallback(const AnyEventCallback& cb){
        _server_closed_callback = cb;
    }
    void Established() // 连接就绪之后，进行channel回调设置
    {
        _loop->RunInLoop(std::bind(&Connnection::EstablishedInLoop,this));
    }
    void Send(const void* data,size_t len) {
        // 外界传入的data，可能是一个临时空间，我们现在只是把发送操作压入了任务池，有可能并没有立即执行
        // 因此有可能执行的时候，data指向的空间就已经被释放了
        Buffer buf;
        buf.WriteAndPush(data,len);
        _loop->RunInLoop(std::bind(&Connnection::SendInLoop,this,std::move(buf)));
    }
    void Shutdown() // 关闭接口，只进行判断有无数据待处理
    {
        _loop->RunInLoop(std::bind(&Connnection::ShutdownInLoop,this));
    }
    void Release() {
        _loop->QueueInLoop(std::bind(&Connnection::RealseInLoop,this));
    }
    void EnableInactiveRealse(int sec) // 启动非活跃销毁
    {
        _loop->RunInLoop(std::bind(&Connnection::EnableInactiveRealseInLoop,this,sec));
    }
    void CancelInactiveRealse() // 取消非活跃销毁
    {
        _loop->RunInLoop(std::bind(&Connnection::CancelInactiveRealseInLoop,this));
    }
    void Upgrade(const ConnectedCallback& conn,const MessageCallback& msg,const ClosedCallback& closed,
                const AnyEventCallback& event,const Any &context) // 切换协议，重置上下文以及阶段性处理函数，非线程安全的-- 这个接口必须在EventLoop线程中立即执行
                // 防备事件触发后，处理的时候，切换任务还没有被执行 -- 会导致诗句使用原协议处理了
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connnection::UpgradeInLoop,this,conn,msg,closed,event,context));
    }
};


class Acceptor{
private:
    Socket _socket; // 创建监听socket
    EventLoop* _loop; // 对监听套接字进行监控
    Channel _channel; // 监听socket事件管理
    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback;
private:
    void HandleRead() // 监听socket读事件回调，调用_accept_callback
    {
        // DBG_LOG("ACCEPTOR HANDLE READ");
        int newfd = _socket.Accept();
        if(newfd < 0){
            return;
        }
        if(_accept_callback) _accept_callback(newfd);
    }
    int CreateServer(int port) {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }
public:
// 不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动
// 否则有可能启动监控后，立即有事件，处理的时候，回调函数还没有设置，新连接得不到处理，会有资源泄露
    Acceptor(EventLoop* loop,int port):_loop(loop),_socket(Acceptor::CreateServer(port)),_channel(_loop,_socket.Fd())
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
    }
    ~Acceptor() = default;
    void SetAcceptCallback(const AcceptCallback& cb) {
        _accept_callback = cb;
    }
    void Listen() {
        _channel.EnableRead();
    }
};


void Channel::Remove() {_loop->RemoveEvent(this); }
void Channel::Update() {_loop->UpdateEvent(this); }
void TimerWheel::TimerAdd(uint64_t id,uint32_t timeout,const TaskFunc& callback){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,timeout,callback));
}
// mod 刷新定时任务
void TimerWheel::TimerRefresh(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}
void TimerWheel::TimerCancel(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}

class TcpServer{
private:
    uint64_t _next_id; // 自增长的连接id
    int _port;
    EventLoop _baseloop; // 主线程的EventLoop对象，负责监听事件
    int _timeout; // 非活跃连接的统计时间
    bool _enable_inactive_release; // 是否启动了非活跃链接超时销毁的连接操作
    Acceptor _acceptor; // 监听套接字管理对象
    LoopThreadPool _pool; // 从属EventLoop线程池
    std::unordered_map<uint64_t,PtrConnection> _conns; // 保存所有连接对应的shared_ptr对象

    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&,Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    using Functor = std::function<void()>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
private:
    void NewConnection(int fd) //为新连接构建一个Connection进行管理
    {
        // DBG_LOG("new connection function");
        PtrConnection conn(new Connnection(_pool.NextLoop(),_next_id,fd));
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetSvrClosedCallback(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
        if(_enable_inactive_release)
            conn->EnableInactiveRealse(_timeout); // 非活跃超时销毁
        conn->Established(); // 就绪初始化
        _conns.insert({_next_id,conn});
        _next_id++;
    }
    void RemoveConnectionInLoop(const PtrConnection& conn){
        int id = conn->Id();
        if(_conns.find(id) != _conns.end())
            _conns.erase(id);
    }
    void RemoveConnection(const PtrConnection& conn) // 从管理Connection中的_conns中移除连接信息
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
    }
    void RunAfterInLoop(const Functor& task,int delay){
        _next_id++;
        _baseloop.TimerAdd(_next_id,delay,task);
    }
public:
    TcpServer(int port):_port(port),_next_id(0),_enable_inactive_release(false),_acceptor(&_baseloop,port),_pool(&_baseloop)
                        {
                            
                            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
                            _acceptor.Listen(); // 将套接字挂载到baseloop上
                        }
    void SetThreadCount(int count){
        return _pool.SetThreadCount(count);
    }
    void SetConnectedCallback(const ConnectedCallback& cb){
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb){
        _message_callback=  cb;
    }
    void SetClosedCallback(const ClosedCallback& cb) {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback& cb){
        _event_callback = cb;
    }
    void EnableInactiveRelease(int timeout){
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    void RunAfter(const Functor& task,int delay) // 用户添加一个定时任务
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
    }
    void Start(){
        _pool.Create(); // 创建从属线程
        return _baseloop.Start();
    }
};

class NetWork{
public:
    NetWork(){
        DBG_LOG("SIGPIPE INIT");
        signal(SIGPIPE,SIG_IGN);
    }
};

static NetWork nw;

#endif